// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: donggu
#pragma once
#include <fmt/format.h>

#include <algorithm>
#include <boost/algorithm/string/predicate.hpp>
#include <ctime>  // strptime
#include <experimental/filesystem>
#include <memory>
#include <regex>
#include <set>
#include <string>
#include <utility>
#include <vector>

#include "cc/coin2/feed/fastfeed/reader.h"
#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/config/app_config.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/base/symbology/product.h"
#include "coin2/exchange/factory/symbology/product.h"
#include "coin2/feed/fastfeed/const.h"
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/record/simple.h"

namespace impl {
using coin2::exchange::symbology::ProductRequest;
using fastfeed::proto::Feed;

inline bool cmp(const std::experimental::filesystem::path& lhs,
         const std::experimental::filesystem::path& rhs) {
  if (lhs.parent_path() != rhs.parent_path()) {
    return lhs.parent_path().parent_path().filename()
        < rhs.parent_path().parent_path().filename();
  } else {
    std::string lhs_file_name_core =
        lhs.filename().string().substr(0, lhs.filename().string().rfind(".W"));
    std::string rhs_file_name_core =
        rhs.filename().string().substr(0, rhs.filename().string().rfind(".W"));
    return lhs_file_name_core < rhs_file_name_core;
  }
}

struct product_less {
  inline bool operator()(
      const std::unique_ptr<IProduct>& lhs,
      const std::unique_ptr<IProduct>& rhs) const {
    return lhs->key() < rhs->key();
  }
};

class FastFeedReader : public orio::io::Reader<Feed> {
 public:
  FastFeedReader(const MarketExchangeApiRecipe& mear,
                 const ProductConfig& config,
                 const std::string& machine,
                 const std::string& root_dir, int worker, int worker2,
                 int64_t warmup, int64_t begin_timestamp, /* inclusive */
                 int64_t end_timestamp,                   /* exclusive */
                 size_t max_record_size)
                 : mear(mear),
                   config(config),
                   machine(machine),
                   root_dir(root_dir),
                   worker(worker),
                   worker2(worker2),
                   warmup(warmup),
                   begin_timestamp(begin_timestamp),
                   end_timestamp(end_timestamp),
                   max_record_size(max_record_size) {
    auto products = ProductRequestsFromConfig(mear.mea.String(), config, begin_timestamp);
    for (auto& prod : products) {
      // std::move, thus ptr is maintained
      products_read_.insert(prod.get());
      products_.insert(std::move(prod));
    }
  }

  const MarketExchangeApiRecipe& mear;
  const ProductConfig& config;
  const std::string& machine;
  const std::string& root_dir;
  int worker;
  int worker2;
  int64_t warmup;
  int64_t begin_timestamp;
  int64_t end_timestamp;
  size_t max_record_size;

  bool BanProduct(const IProduct* product) {
    if (products_read_.count(product) > 0) {
      LOG(INFO) << "Unsubscribed " << product->relative_norm();
      products_read_.erase(product);
    }
    return products_read_.count(product) == 0;
  }

  void LazyInitializeReader() {
    std::vector<std::unique_ptr<orio::io::Reader<Feed>>> readers;
    for (auto& prod : products_read_) {
      LOG(INFO) << prod->relative_norm() << ": " << prod->absolute_norm();

      std::vector<std::experimental::filesystem::path> paths =
          GetPathList(machine, root_dir, prod, mear, worker, worker2, warmup,
                      begin_timestamp, end_timestamp);

      // 20191010/BTC.gz --
      // 20191011/BTC.gz --   -> MultiFileReader -> SimpleReader ->
      // FastFeedProto Reader (max_size)
      // -> TimeFilter (IntervalFastFeedReader)

      // 20191010/ETH.gz --
      // 20191011/ETH.gz --   -> MultiFileReader -> SimpleReader ->
      // FastFeedProto Reader -> IntervalFastFeedReader

      // Merge(BTC, ETH) by timestamp

      if (paths.empty()) {
        LOG(WARNING) << "paths are empty!!";
        continue;
      }
      std::sort(std::begin(paths), std::end(paths), cmp);
      for (auto& path : paths) {
        LOG(INFO) << "[FastFeed Reader] will read " << path;
      }

      if (mear.mea.market == Options) {
        for (auto& path : paths) {
          readers.push_back(
              std::unique_ptr<orio::io::Reader<Feed>>(new IntervalFastFeedReader(
                  std::unique_ptr<orio::io::Reader<Feed>>(new FastFeedProtoReader(
                      std::unique_ptr<orio::record::SimpleReader>(
                          orio::record::SimpleReader::FromFile(path, max_record_size)),
                      max_record_size)),
                  begin_timestamp - warmup, end_timestamp)));
        }
      } else {
        readers.push_back(
            std::unique_ptr<orio::io::Reader<Feed>>(new IntervalFastFeedReader(
                std::unique_ptr<orio::io::Reader<Feed>>(new FastFeedProtoReader(
                    std::unique_ptr<orio::record::SimpleMultiFileReader>(
                        orio::record::SimpleMultiFileReader::FromFile(paths, max_record_size)),
                    max_record_size)),
                begin_timestamp - warmup, end_timestamp)));
      }
    }
    reader_ = std::make_unique<MergedFastFeedReader>(std::move(readers));
  }

  bool Read(Feed* out) override {
    if (!reader_) {
      LazyInitializeReader();
    }
    return reader_->Read(out);
  }

  std::vector<IProduct*> products() {
    std::vector<IProduct*> out;
    for (const auto& p : products_) {
      out.push_back(p.get());
    }
    return out;
  }

  // /remote/iosg/coin-2/feed_cache/
  // feed-02.us-east-1.aws/20190608/Spot.Gdax/ETH-BTC.Gdax.ETH-BTC.W2.gz
  // BTC-USD.PERPETUAL.Bitmex.XBTUSD.W1.gz
  // BCHABC-USD.20191122.Okex.BCH1122.W1.gz
  inline static std::string DEFAULT_ROOT = "/remote/iosg/coin-2/feed_cache/";

 private:
  std::vector<std::experimental::filesystem::path> GetPathList(
      const std::string& machine, const std::string& root_dir,
      const IProduct* product,
      const MarketExchangeApiRecipe& mear,
      int worker, int worker2, int64_t warmup, int64_t begin_timestamp,
      int64_t end_timestamp /* exclusive */) {
    std::vector<std::experimental::filesystem::path> out;
    std::string fastfeed_symbol = product->absolute_norm();

    auto p = PtimeFromTimestamp(begin_timestamp);
    begin_timestamp = TimestampFromPtime(PtimeFromDatetime(
        p.date().year(), p.date().month(), p.date().day(), 0, 0, 0, 0));

    std::vector<std::string> machines = Split(machine, ",");
    for (int64_t ts = begin_timestamp; ts < end_timestamp;
        ts += 24 * 60 * 60 * 1000000000ull) {
      std::vector<std::experimental::filesystem::path> out_each;
      for (auto machine_i : machines) {
        if (!out_each.empty()) {
          break;
        }
        char yyyymmdd[9];
        time_t t = ts / 1'000'000'000ull;
        tm tm_;
        gmtime_r(&t, &tm_);
        std::strftime(yyyymmdd, 9, "%Y%m%d", &tm_);

        std::vector<std::string> path_chosen = [&] {
          for (auto resolve_rule : {COIN21_MAP, MEAR, MEA, MER, ME}) {
            using std::string_literals::operator""s;
            for (const std::string& extension : {".lz4"s, ".gz"s, ""s}) {
              auto resolve_dir = ResolveFastfeedCacheDir(mear, resolve_rule);
              if (resolve_dir.empty()) {
                continue;
              }
              for (int worker_try : {worker, worker2}) {
                auto complete_dir = fmt::format(
                    "{}/{}/{}/{}/", root_dir,
                    ResolveFastfeedMachine(mear, machine_i), yyyymmdd, resolve_dir);
                if (!std::experimental::filesystem::exists(complete_dir)) {
                  if (mear.mea.exchange == ExchangeType::Okex) {
                    if (mear.recipe.find("bbo") != std::string::npos) {
                      boost::replace_first(resolve_dir, "bbo", "realtime");
                      complete_dir =
                          fmt::format("{}/{}/{}/{}/", root_dir,
                                      ResolveFastfeedMachine(mear, machine_i),
                                      yyyymmdd, resolve_dir);
                    }
                    if (!std::experimental::filesystem::exists(complete_dir)) {
                      if (mear.mea.api.find("v5") != std::string::npos) {
                        boost::replace_all(resolve_dir, "v5", "v3");
                        complete_dir =
                            fmt::format("{}/{}/{}/{}/", root_dir,
                                        ResolveFastfeedMachine(mear, machine_i),
                                        yyyymmdd, resolve_dir);
                        if (!std::experimental::filesystem::exists(complete_dir)) {
                          continue;
                        }
                      } else {
                        continue;
                      }
                    }
                  } else if (mear.mea.exchange == ExchangeType::Bybit) {
                    if (mear.recipe.find("bbo") != std::string::npos) {
                      boost::replace_first(resolve_dir, "bbo", "realtime");
                      complete_dir =
                          fmt::format("{}/{}/{}/{}/", root_dir,
                                      ResolveFastfeedMachine(mear, machine_i),
                                      yyyymmdd, resolve_dir);
                    }
                    if (!std::experimental::filesystem::exists(complete_dir)) {
                      if (mear.mea.api.find("v3") != std::string::npos) {
                        if (mear.mea.market == Spot) {
                          boost::replace_all(resolve_dir, "v3", "v1");
                        } else if (mear.mea.market == Futures) {
                          boost::replace_all(resolve_dir, "v3", "v2");
                        }
                        complete_dir =
                            fmt::format("{}/{}/{}/{}/", root_dir,
                                        ResolveFastfeedMachine(mear, machine_i),
                                        yyyymmdd, resolve_dir);
                        if (!std::experimental::filesystem::exists(complete_dir)) {
                          continue;
                        }
                      } else {
                        continue;
                      }
                    }
                  } else {
                    continue;
                  }
                }
                auto worker_tag = fmt::format("W{}", worker_try);
                std::vector<std::string> picked;
                for (auto& path :
                    std::experimental::filesystem::directory_iterator(
                        complete_dir)) {
                  auto filename = path.path().filename().string();
                  std::vector<std::string> splitted;
                  splitted = Split(filename, ".");
                  std::string fastfeed_symbol_to_check = fastfeed_symbol;
                  // Optioins case, try to match all strike prices.
                  if (splitted.size() == 7) {
                    if (fastfeed_symbol.find(".0.") != std::string::npos) {
                      fastfeed_symbol_to_check = fastfeed_symbol.substr(
                        0, fastfeed_symbol.rfind(".0."));
                    }
                    if (filename.find("." + worker_tag) != std::string::npos &&
                        filename.find(extension) != std::string::npos) {
                      if (filename.rfind(fastfeed_symbol_to_check + ".", 0) == 0) {
                        picked.push_back(path.path().string());
                        auto target_dot = filename.find(".", fastfeed_symbol_to_check.size() + 1);
                        auto norm = filename.substr(0, target_dot);
                        ProductRequest req;
                        req.set_mea(mear.mea.String());
                        req.set_norm(norm);
                        auto product = ResolveProductRequest(req, begin_timestamp);
                        products_.insert(std::move(product));
                      }
                    }
                  } else {
                    if (splitted.size() == 5) {
                      auto start_time = splitted[2];
                      auto timestamp = TimestampFromIso8601(start_time);
                      ProductRequest req;
                      req.set_mea(mear.mea.String());
                      req.set_norm(product->relative_norm());
                      auto product = ResolveProductRequest(req, timestamp);
                      fastfeed_symbol_to_check = product->absolute_norm();
                      products_.insert(std::move(product));
                    }
                    if (filename.find("." + worker_tag) != std::string::npos &&
                        filename.find(extension) != std::string::npos) {
                      if (filename.rfind(fastfeed_symbol_to_check + ".", 0) == 0) {
                        picked.push_back(path.path().string());
                      } else if (fastfeed_symbol_to_check.find("BCHN") !=
                                std::string::npos) {
                        // Try BCHABC
                        // TODO(gekim): Remove this hack after finishing the regeneration of BCH.
                        std::string bchabc = std::string("BCHABC").append(
                            fastfeed_symbol_to_check.substr(4));
                        if (filename.rfind(bchabc + ".", 0) == 0) {
                          picked.push_back(path.path().string());
                        }
                      }
                    }
                  }
                }
                if (!picked.empty()) {
                  return picked;
                }
              }
            }
          }
          LOG(ERROR) << "[FastFeedReader] missing file: " << mear.String() << " "
                    << fastfeed_symbol;
          return std::vector<std::string>{};
        }();
        if (!path_chosen.empty()) {
          out_each.insert(std::end(out_each), std::begin(path_chosen),
                    std::end(path_chosen));
        }
      }
      out.insert(std::end(out), std::begin(out_each), std::end(out_each));
    }
    return out;
  }

  std::unique_ptr<MergedFastFeedReader> reader_;
  std::set<std::unique_ptr<IProduct>, product_less> products_;
  std::set<const IProduct *> products_read_;
};

}  // namespace impl

using impl::FastFeedReader;
